package pairs;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 使用pair的方式，使用自定义了TextPiar Writable对象
 *
 */
public class mapper extends Mapper<LongWritable, Text, TextPair, DoubleWritable> {

    @Override
    public void map(LongWritable inputKey, Text inputValue, Context context)
            throws IOException, InterruptedException {

        String doc = inputValue.toString();
        //这里只是简单的根据正则分词，如果希望准确分词，请使用相关分词包
        String reg = "[\\p{P}\\s]";
        String[] allTerms = doc.split(reg);
        for(int i = 0; i < allTerms.length; i++) {
            if((!"".equals(allTerms[i])) && allTerms[i] != null) {
                //考虑in-mapper combining
                Map<String, Integer> pairMap = new HashMap<String, Integer>();

                //取出该单词对应的一定窗口大小内的共现词
                String[] termNeighbors = neighborsOfTerm(allTerms[i], i, allTerms, 3);
                for(String nbTerm : termNeighbors) {
                    if((!"".equals(nbTerm)) && nbTerm != null) {
                        String textPairStr = allTerms[i] + "," + nbTerm;
                        //in-mapper combining
                        if(!pairMap.containsKey(textPairStr)) {
                            pairMap.put(textPairStr, 1);
                        } else {
                            pairMap.put(textPairStr, pairMap.get(textPairStr) + 1);
                        }

                    }
                }
                for(Map.Entry<String, Integer> entry: pairMap.entrySet()) {
                    String[] pairStrs = entry.getKey().split(",");
                    TextPair textPair = new TextPair(pairStrs[0], pairStrs[1]);
                    context.write(textPair, new DoubleWritable(entry.getValue()));
                }
            }

        }

    }

    /**
     * 计算某个词在某窗口大小内的共现词
     * @param term
     * @param allterms
     * @return
     */
    public String[] neighborsOfTerm(String term, int pos, String[] allterms, int windowSize) {
        String[] neighbors = new String[windowSize];
        int count = allterms.length;
        int j = 0;
        int leftOffSet = 0;
        int rightOffSet = 0;
        if(pos < windowSize / 2) {
            leftOffSet = pos;
            rightOffSet = windowSize - leftOffSet;
        } else if (pos >= count - 1 - windowSize / 2) {
            rightOffSet = count - 1 - pos;
            leftOffSet = windowSize - rightOffSet;
        } else {
            leftOffSet = windowSize / 2;
            rightOffSet = windowSize - leftOffSet;
        }
        for(int i = pos - leftOffSet; i <= pos + rightOffSet && i >=0 && i < count; i++) {
            if(term != allterms[i] ) {
                neighbors[j] = allterms[i];
                j ++;
            }
        }

        return neighbors;
    }
}
